package mrunit;

import com.ailk.oci.ocnosql.common.rowkeygenerator.GenRKCallBackHBImpl;
import com.ailk.oci.ocnosql.common.rowkeygenerator.MD5RowKeyGenerator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.IdentityTableReducer;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by zhaopan on 2017/6/29.<br>
 *
 * @Description:
 */
public class PutDate
{
    public static class TokenizerMapper
            extends Mapper<LongWritable, Text, Text, Put>
    {
        private Text word = new Text();
        private final static List<String> columns = new ArrayList<String>();

        static
        {
            columns.add("BTIME");
            columns.add("ETIME");
            columns.add("PROTOCOLID");
            columns.add("EVENTID");
            columns.add("SPCKIND");
            columns.add("MSCCODE");
            columns.add("BSCCODE");
            columns.add("LAC");
            columns.add("CI");
            columns.add("OLAC");
            columns.add("OCI");
            columns.add("DLAC");
            columns.add("DCI");
            columns.add("FIRSTLAC");
            columns.add("FIRSTCI");
            columns.add("LASTLAC");
            columns.add("LASTCI");
            columns.add("DRNCID");
            columns.add("CALLINGNUM");
            columns.add("CALLEDNUM");
            columns.add("CALLINGIMSI");
            columns.add("CALLEDIMSI");
            columns.add("CALLINGIMEI");
            columns.add("CALLEDIMEI");
            columns.add("EVENTRESULT");
            columns.add("HANDOUTREQUIRECAUSE");
            columns.add("HANDOUTOFFSET");
            columns.add("HANDOUTRSPOFFSET");
            columns.add("HANDINOFFSET");
            columns.add("HANDINRSPOFFSET");
            columns.add("HANDOUTSTATUS");
            columns.add("HANDINSTATUS");
            columns.add("LUSTATUS");
            columns.add("HOFLAG");
            columns.add("ALERT_TIME");
            columns.add("CONN_TIME");
            columns.add("CALL_DUR");
        }

        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException
        {
            String line = value.toString();
            String[] wordArray = line.split(",");
            StringBuffer sb = new StringBuffer();
            if (wordArray.length == 37)
            {
                String phone = wordArray[18];
                String beginTime = wordArray[0];
                if (phone != null && beginTime != null && !"".equals(phone) && !"".equals(beginTime))
                {
                    //System.out.println("phone:" + phone);
                    String md5 = new MD5RowKeyGenerator().generatePrefix(phone).toString();
                    String rowKey = new GenRKCallBackHBImpl().callback(md5 + "" + phone + "" + beginTime, line, null);//
                    word.set(rowKey);
                    //System.out.println("rowKey" + rowKey);
                    byte[] newrowKey = Bytes.toBytes(rowKey);
                    // ImmutableBytesWritable rowKeyWritable = new ImmutableBytesWritable(newrowKey);
                    byte[] family = Bytes.toBytes("0");
                    Put put = new Put(newrowKey);
                    for (int i = 0; i < columns.size(); i++)
                    {
                        byte[] qualifier = Bytes.toBytes(columns.get(i));
                        byte[] hbaseValue = Bytes.toBytes(wordArray[i]);
                        put.addColumn(family, qualifier, hbaseValue);
                    }
                    context.write(word, put);
                }
            }
        }
    }

    public static void main(String[] args) throws Exception
    {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.security.authentication","kerberos");
        conf.set("hbase.zookeeper.quorum","hbbdch-dn-01:2181,hbbdch-nn-01:2181,hbbdch-nn-02:2181");
        conf.set("zookeeper.znode.parent","/services/slider/users/client/hbase_1007_3");
        conf.set("hbase.master.kerberos.principal","client/_HOST@HBBDCH.COM");
        conf.set("hbase.regionserver.kerberos.principal", "client/_HOST@HBBDCH.COM");
        conf.setInt("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", 5000);
        conf.set("hbase.fs.tmp.dir", "/tmp/hbase-staging");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        Job job = Job.getInstance(conf, "hdfs to hbase");
        job.setJarByClass(PutDate.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Put.class);//important
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        TableMapReduceUtil.initTableReducerJob(otherArgs[1], IdentityTableReducer.class, job);
        job.setNumReduceTasks(0);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}